package com.hk.commons.locks.zookeeper;

import com.hk.commons.util.Lazy;
import com.hk.commons.util.SpringContextHolder;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessMultiLock;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * <h2>基于zookeeper实现分布式锁</h2>
 * <p>实现原理：</p>
 * <pre>
 * 基于唯一节点特性来实现分布式锁的操作，多个应用程序去抢占锁资源时，只需要在指定节点上创建一个节点(如 /lock)，
 *      由于Zookeeper中节点的唯一性特性，使得只会有一个用户成功创建节点，剩下没有创建成功的用户表示竞争锁失败。
 *      这种方法虽然能达到目的，但是会有一个问题:
 *          假设有非常多的节点需要等待获得锁，那么等待的方式自然是使用watcher机制来监听节点的删除事件，
 *          一旦发现该节点被删除说明之前获得锁的节点已经释放了锁，那么此时剩下的所有节点会同时收到删除事件从而去竞争锁，这个过程会产生惊群效应。
 *
 * 什么是 "惊群效应" 呢？
 *      简单来说就是如果存在许多的客户端在等待获取锁，当成功获取到锁的进程释放该节点后，所有处于等待状态的客户端都会被唤醒，
 *      这个时候zookeeper会在短时间内发送大量子节点变更事件给所有待获取锁的客户端，然后实际情况是只会有一个客户端获得锁。
 *      如果在集群规模比较大的情况下，会对zookeeper服务器的性能产生比较的影响。
 *
 * 解决 "惊群效应" 方法：采用Zookeeper的有序节点特性来实现分布式锁
 *      每个客户端都往指定的节点下注册一个临时有序节点，越早创建的节点，节点的顺序编号就越小，那么我们可以判断子节点中最小的节点设置为获得锁。
 *      如果自己的节点不是所有子节点中最小的，意味着还没有获得锁。这个的实现和前面单节点实现的差异性在于，每个节点只需要监听比自己小的节点，
 *      当比自己小的节点删除以后，客户端会收到watcher事件，此时再次判断自己的节点是不是所有子节点中最小的，如果是则获得锁，否则就不断重复这个过程，
 *      这样就不会导致惊群效应，因为每个客户端只需要监控一个节点。
 * </pre>
 * <p>
 * 比如：锁的节点为 /lock
 * <pre> {@code
 *      ZookeeperLock lock = new ZookeeperLock("/lock");
 *      // 调用  lock()方法时，会竞争锁，如果第一个线程获取到了锁，第二个线程获取时，会在 /lock/下创建子节点(这些创建的子节点是可以排序的)，如: _c_0ce95645-59aa-42a5-86bc-eb4a452de0fb-lock-0000000001,
 *      // 这个节点会watch 第一个线程释放锁的事件，同理，第三个线程获取时，会在 /lock/下创建子节点，如: _c_0ce95645-59aa-42a5-86bc-eb4a452de0fb-lock-0000000002,
 *      // 这个节点会watch 第二个线程释放锁的事件，依此类推，每个后面的线程，会watch前面线程的释放锁事件，就不会导致惊群效应
 *      lock.lock();
 *      try{
 *          Do something...
 *      }finally{
 *          lock.unlock();
 *      }
 * }</pre>
 * <pre>
 * 这里默认使用 {@link InterProcessMutex} 实现zookeeper分布式锁，
 * 当然，还有 {@link  #multiLock(List)} 和 {@link  #semaphoreMutexLock(String)} 分布式排它锁，
 * 以及{@link org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock} 分布式读写锁
 * </pre>
 *
 * @author Kevin
 * @date 2022-08-30 20:20:28
 */
@Slf4j
public class ZookeeperLock implements Lock {

    private final static Lazy<CuratorFramework> CURATOR_FRAMEWORK;

    static {
        CURATOR_FRAMEWORK = Lazy.of(() -> SpringContextHolder.getBean(CuratorFramework.class));
    }

    public final InterProcessLock lock;

    /**
     * 分布式可重入排它锁
     */
    public ZookeeperLock(String path) {
        this.lock = new InterProcessMutex(CURATOR_FRAMEWORK.get(), path);
    }

    private ZookeeperLock(InterProcessLock lock) {
        this.lock = lock;
    }

    /**
     * 支持多个锁同时上锁、解锁
     */
    public static ZookeeperLock multiLock(List<String> path) {
        return new ZookeeperLock(new InterProcessMultiLock(CURATOR_FRAMEWORK.get(), path));
    }

    /**
     * 分布式排它锁
     */
    public static ZookeeperLock semaphoreMutexLock(String path) {
        return new ZookeeperLock(new InterProcessSemaphoreMutex(CURATOR_FRAMEWORK.get(), path));
    }

    @Override
    @SneakyThrows
    public void lock() {
        try {
            lock.acquire();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            lock();
        }
    }

    @Override
    @SneakyThrows
    public void lockInterruptibly() {
        try {
            lock.acquire();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw e;
        }
    }

    @Override
    @SneakyThrows(value = Exception.class)
    public boolean tryLock() {
        return lock.acquire(0, null);
    }

    @Override
    @SneakyThrows
    public boolean tryLock(long time, TimeUnit unit) {
        try {
            return lock.acquire(time, unit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw e;
        }
    }

    @Override
    @SneakyThrows
    public void unlock() {
        lock.release();
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException("newCondition UnSupported.");
    }
}
